package com.lry.block;

import com.google.common.primitives.Ints;
import com.lry.util.DynamicSliceOutput;
import com.lry.util.IntVector;
import com.lry.util.Slice;
import com.lry.util.VariableLengthQuantity;

import java.util.Comparator;

import static com.google.common.base.Preconditions.*;
import static com.lry.util.SizeOf.SIZE_OF_INT;
import static java.util.Objects.requireNonNull;

public class BlockBuilder {


    private final int blockRestartInterval;

    private final IntVector restartPositions;

    private final Comparator<Slice> comparator;

    private int entryCount;

    private int restartBlockEntryCount;

    private boolean finished;

    private final DynamicSliceOutput block;

    private Slice lastKey;

    public BlockBuilder(int estimatedSize, int blockRestartInterval, Comparator<Slice> comparator){
        checkArgument(estimatedSize >= 0, "estimatedSize is negative");
        checkArgument(blockRestartInterval >= 0, "blockRestartInterval is negative");
        requireNonNull(comparator, "comparator is null");

        this.block = new DynamicSliceOutput(estimatedSize);

        this.blockRestartInterval = blockRestartInterval;

        this.comparator = comparator;

        this.restartPositions = new IntVector(32);
        //第一组都是从0开始
        this.restartPositions.add(0);
    }

    public void reset(){
        block.reset();
        entryCount = 0;
        restartPositions.clear();
        restartPositions.add(0); // first restart point must be 0
        restartBlockEntryCount = 0;
        lastKey = null;
        finished = false;
    }

    public int getEntryCount()
    {
        return entryCount;
    }

    public boolean isEmpty()
    {
        return entryCount == 0;
    }


    public int currentSizeEstimate(){
        if(finished){
            return block.size();
        }
        // no records is just a single int
        if (block.size() == 0) {
            return SIZE_OF_INT;
        }

        return block.size() +                              // raw data buffer
                restartPositions.size() * SIZE_OF_INT +    // restart positions
                SIZE_OF_INT;
    }


    public void add(BlockEntry blockEntry){
        requireNonNull(blockEntry, "blockEntry is null");

        add(blockEntry.getKey(),blockEntry.getValue());
    }

    public void add(Slice key, Slice value) {
        requireNonNull(key, "key is null");
        requireNonNull(value, "value is null");
        checkState(!finished, "block is finished");
        checkPositionIndex(restartBlockEntryCount, blockRestartInterval);

        checkArgument(lastKey == null || comparator.compare(key, lastKey) > 0, "key must be greater than last key");

        int sharedKeyBytes = 0;
        //还在本组内，计算共享key的长度
        if(restartBlockEntryCount < blockRestartInterval){
            sharedKeyBytes = calculateSharedBytes(key, lastKey);
        }else {
            //另开一组, 记录新组指向block entry的起始offset
            restartPositions.add(block.size());
            restartBlockEntryCount = 0;
        }

        int nonSharedKeyBytes = key.length() - sharedKeyBytes;

        // write "<shared><non_shared><value_size>"
        VariableLengthQuantity.writeVariableLengthInt(sharedKeyBytes, block);
        VariableLengthQuantity.writeVariableLengthInt(nonSharedKeyBytes, block);
        VariableLengthQuantity.writeVariableLengthInt(value.length(), block);


        // write non-shared key bytes
        block.writeBytes(key, sharedKeyBytes, nonSharedKeyBytes);

        // write value bytes
        block.writeBytes(value, 0, value.length());

        // update last key
        lastKey = key;

        // update state
        entryCount++;
        restartBlockEntryCount++;
    }

    public static int calculateSharedBytes(Slice leftKey,Slice rightKey){
        int sharedKeyBytes = 0;

        if(null!=leftKey && null!=rightKey){
            int minSharedKeyBytes = Ints.min(leftKey.length(), rightKey.length());
            while (sharedKeyBytes < minSharedKeyBytes && leftKey.getByte(sharedKeyBytes) == rightKey.getByte(sharedKeyBytes)) {
                sharedKeyBytes++;
            }
        }
        return sharedKeyBytes;
    }

    public Slice finish(){
        if(!finished){
            finished = true;

            if(entryCount>0){
                //往block中添加每个group的entry起始偏移位置
                restartPositions.write(block);
                //group len
                block.writeInt(restartPositions.size());
            }else{
                //group len=0
                block.writeInt(0);
            }
        }
        return block.slice();
    }
}
